package com.atguigu.day04;

import com.atguigu.utils.IntegerSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

// 使用KeyedProcessFunction实现求平均值的需求
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntegerSource())
                .keyBy(r -> 1)
                .process(new Avg())
                .print();

        env.execute();
    }

    public static class Avg extends KeyedProcessFunction<Integer, Integer, Integer> {
        // 使用值状态变量实现累加器
        // 声明值状态变量
        private ValueState<Tuple2<Integer, Integer>> accumulator;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 初始化值状态变量
            // accumulator不可能为空
            // accumulator.value()可能为空
            accumulator = getRuntimeContext().getState(
                    // 第一个参数：状态变量的名字
                    // 第二个参数：状态变量的类型
                    new ValueStateDescriptor<Tuple2<Integer, Integer>>("accumulator", Types.TUPLE(Types.INT, Types.INT))
            );
        }

        @Override
        public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
            if (accumulator.value() == null) {
                // 如果状态变量里面为null，说明第一条数据到达
                accumulator.update(Tuple2.of(value, 1));
            } else {
                // 读取状态变量的值
                Tuple2<Integer, Integer> acc = accumulator.value();
                // 写回状态变量
                // 更新完累加器，value被丢弃
                accumulator.update(Tuple2.of(acc.f0 + value, acc.f1 + 1));
            }

            // 输出平均值
            out.collect(accumulator.value().f0 / accumulator.value().f1);
        }
    }
}
